# -*- coding:utf-8 -*-

import asyncio
import asyncssh
from asyncssh import PermissionDenied, ConnectionLost, Error as AsyncSSHError
import os
import time
import base64
import shlex
import socket
from datetime import datetime
from typing import List, Optional, Dict, Any, Tuple
from sqlalchemy.ext.asyncio import AsyncSession
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import tarfile
import io

from module_admin.dao.server_host_dao import ServerHostDao, ServerCommandLogDao
from module_admin.entity.do.server_host_do import ServerHost, ServerCommandLog
from module_admin.entity.vo.server_host_vo import (
    ServerHostModel,
    ServerHostQueryModel,
    ServerHostCreateModel,
    ServerHostUpdateModel,
    ServerHostBatchImportModel,
    CommandExecuteModel,
    CommandExecuteResultModel,
    AgentDeployModel,
    AgentControlModel,
    AgentLogQueryModel
)
from utils.page_util import PageResponseModel
from utils.log_util import logger

# ✅ 在模块级别导入，避免在嵌套函数中导入导致的上下文问题
from config.database import AsyncSessionLocal


class PasswordEncryptor:
    """密码加密工具"""
    
    # 使用环境变量或配置文件中的密钥，这里使用示例密钥
    # 在生产环境中应该使用环境变量
    _SECRET_KEY = "flux-panel-server-host-secret-key-2025"
    
    @classmethod
    def _get_cipher(cls) -> Fernet:
        """获取加密器"""
        # 使用PBKDF2HMAC从密钥派生Fernet密钥
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=b'flux-panel-salt',
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(cls._SECRET_KEY.encode()))
        return Fernet(key)
    
    @classmethod
    def encrypt_password(cls, password: str) -> str:
        """加密密码"""
        cipher = cls._get_cipher()
        encrypted = cipher.encrypt(password.encode())
        return encrypted.decode()
    
    @classmethod
    def decrypt_password(cls, encrypted_password: str) -> str:
        """解密密码"""
        cipher = cls._get_cipher()
        decrypted = cipher.decrypt(encrypted_password.encode())
        return decrypted.decode()


class SSHConnectionManager:
    """SSH连接管理器"""
    
    @staticmethod
    async def create_connection(
        host_ip: str,
        ssh_port: int,
        ssh_username: str,
        ssh_password: str,
        timeout: int = 10
    ) -> Optional[asyncssh.SSHClientConnection]:
        """创建SSH连接"""
        # 临时禁用SSH密钥查找，避免访问 /root/.ssh/crt 等文件
        original_home = os.environ.get('HOME')
        original_userprofile = os.environ.get('USERPROFILE')
        
        try:
            # 临时设置一个不存在的HOME目录，阻止asyncssh查找密钥文件
            # 使用临时目录而不是 /root/.ssh
            temp_ssh_dir = os.path.join(os.path.expanduser('~'), '.ssh_temp_disabled')
            os.environ['HOME'] = temp_ssh_dir
            if 'USERPROFILE' in os.environ:
                os.environ['USERPROFILE'] = temp_ssh_dir
            
            try:
                conn = await asyncio.wait_for(
                    asyncssh.connect(
                        host_ip,
                        port=ssh_port,
                        username=ssh_username,
                        password=ssh_password,
                        known_hosts=None,  # 不验证known_hosts
                        client_keys=[],  # 不使用密钥认证（空列表）
                        preferred_auth=('password', 'keyboard-interactive')  # 只使用密码认证
                    ),
                    timeout=timeout
                )
                return conn
            except PermissionDenied as e:
                error_msg = f"SSH认证失败: 用户名或密码错误"
                logger.error(f"SSH连接失败 {host_ip}:{ssh_port} - {error_msg}")
                raise Exception(f"{error_msg}\n请检查：\n1. SSH用户名是否正确（当前: {ssh_username}）\n2. SSH密码是否正确\n3. 服务器是否允许密码登录（检查 /etc/ssh/sshd_config 中的 PasswordAuthentication）")
            except ConnectionLost as e:
                error_msg = f"SSH连接中断: 网络不稳定或服务器关闭连接"
                logger.error(f"SSH连接失败 {host_ip}:{ssh_port} - {error_msg}")
                raise Exception(f"{error_msg}\n请检查：\n1. 网络连接是否正常\n2. 服务器是否正常运行\n3. 防火墙是否允许SSH连接")
            except asyncio.TimeoutError:
                error_msg = f"SSH连接超时: 服务器在{timeout}秒内无响应"
                logger.error(f"SSH连接失败 {host_ip}:{ssh_port} - {error_msg}")
                raise Exception(f"{error_msg}\n请检查：\n1. 服务器IP地址是否正确（当前: {host_ip}）\n2. SSH端口是否正确（当前: {ssh_port}）\n3. 防火墙是否开放SSH端口\n4. 服务器是否正常运行")
            except AsyncSSHError as e:
                error_msg = f"SSH连接错误: {str(e)}"
                logger.error(f"SSH连接失败 {host_ip}:{ssh_port} - {error_msg}")
                raise Exception(f"{error_msg}\n请检查：\n1. 服务器SSH服务是否正常运行\n2. 网络连接是否正常\n3. 防火墙配置是否正确")
            except Exception as e:
                error_msg = f"SSH连接失败: {str(e)}"
                logger.error(f"SSH连接失败 {host_ip}:{ssh_port} - {error_msg}")
                raise Exception(f"{error_msg}\n请检查：\n1. 服务器信息是否正确\n2. 网络连接是否正常\n3. 服务器SSH服务是否正常运行")
        finally:
            # 恢复原始环境变量
            if original_home:
                os.environ['HOME'] = original_home
            elif 'HOME' in os.environ:
                del os.environ['HOME']
            
            if original_userprofile:
                os.environ['USERPROFILE'] = original_userprofile
            elif 'USERPROFILE' in os.environ:
                del os.environ['USERPROFILE']
    
    @staticmethod
    async def execute_command(
        conn: asyncssh.SSHClientConnection,
        command: str,
        timeout: int = 300  # 增加到300秒（5分钟），适应长时间命令
    ) -> Tuple[str, str, int]:
        """执行命令"""
        try:
            result = await asyncio.wait_for(
                conn.run(command),
                timeout=timeout
            )
            return result.stdout, result.stderr, result.exit_status
        except asyncio.TimeoutError:
            raise Exception(f"命令执行超时（{timeout}秒）")
        except Exception as e:
            logger.error(f"命令执行失败: {str(e)}")
            raise Exception(f"命令执行失败: {str(e)}")
    
    @staticmethod
    async def upload_file(
        conn: asyncssh.SSHClientConnection,
        local_path: str,
        remote_path: str
    ) -> bool:
        """上传文件"""
        try:
            async with conn.start_sftp_client() as sftp:
                await sftp.put(local_path, remote_path)
            return True
        except Exception as e:
            logger.error(f"文件上传失败: {str(e)}")
            raise Exception(f"文件上传失败: {str(e)}")
    
    @staticmethod
    async def download_file(
        conn: asyncssh.SSHClientConnection,
        remote_path: str,
        local_path: str
    ) -> bool:
        """下载文件"""
        try:
            async with conn.start_sftp_client() as sftp:
                await sftp.get(remote_path, local_path)
            return True
        except Exception as e:
            logger.error(f"文件下载失败: {str(e)}")
            raise Exception(f"文件下载失败: {str(e)}")
    
    @staticmethod
    async def read_file(
        conn: asyncssh.SSHClientConnection,
        remote_path: str
    ) -> str:
        """读取远程文件内容"""
        try:
            async with conn.start_sftp_client() as sftp:
                async with sftp.open(remote_path, 'r') as f:
                    content = await f.read()
                    return content.decode('utf-8') if isinstance(content, bytes) else content
        except Exception as e:
            logger.error(f"文件读取失败: {str(e)}")
            raise Exception(f"文件读取失败: {str(e)}")


class ServerHostService:
    """服务器主机服务"""

    @staticmethod
    async def get_server_host_list(
        db: AsyncSession,
        query: ServerHostQueryModel,
        page_num: int,
        page_size: int
    ) -> PageResponseModel:
        """获取主机列表（分页）"""
        query_dict = query.model_dump(exclude_none=True)
        offset = (page_num - 1) * page_size
        
        hosts, total = await ServerHostDao.get_server_host_list(
            db, query_dict, offset, page_size
        )
        
        # 转换为VO模型
        host_list = [ServerHostModel.model_validate(host) for host in hosts]
        
        return PageResponseModel(
            rows=host_list,
            total=total
        )

    @staticmethod
    async def get_server_host_detail(db: AsyncSession, host_id: int) -> Optional[ServerHostModel]:
        """获取主机详情"""
        host = await ServerHostDao.get_server_host_by_id(db, host_id)
        if host:
            return ServerHostModel.model_validate(host)
        return None

    @staticmethod
    async def create_server_host(
        db: AsyncSession,
        create_data: ServerHostCreateModel,
        create_by: int,
        dept_id: int
    ) -> ServerHost:
        """创建主机"""
        # 检查IP是否已存在
        existing_host = await ServerHostDao.get_server_host_by_ip(db, create_data.host_ip)
        if existing_host:
            raise Exception(f"主机IP {create_data.host_ip} 已存在")
        
        # 加密密码
        encrypted_password = PasswordEncryptor.encrypt_password(create_data.ssh_password)
        
        # 创建主机对象
        host = ServerHost(
            host_name=create_data.host_name,
            host_ip=create_data.host_ip,
            ssh_port=create_data.ssh_port,
            ssh_username=create_data.ssh_username,
            ssh_password=encrypted_password,
            agent_path=create_data.agent_path,
            remark=create_data.remark,
            status=create_data.status,
            create_by=create_by,
            dept_id=dept_id,
            create_time=datetime.now(),
            update_time=datetime.now()
        )
        
        return await ServerHostDao.create_server_host(db, host)

    @staticmethod
    async def update_server_host(
        db: AsyncSession,
        host_id: int,
        update_data: ServerHostUpdateModel
    ) -> bool:
        """更新主机信息"""
        update_dict = update_data.model_dump(exclude_none=True)
        
        # 如果更新密码，需要加密
        if 'ssh_password' in update_dict and update_dict['ssh_password']:
            update_dict['ssh_password'] = PasswordEncryptor.encrypt_password(
                update_dict['ssh_password']
            )
        
        update_dict['update_time'] = datetime.now()
        
        return await ServerHostDao.update_server_host(db, host_id, update_dict)

    @staticmethod
    async def delete_server_host(db: AsyncSession, host_id: int) -> bool:
        """删除主机"""
        return await ServerHostDao.delete_server_host(db, host_id)

    @staticmethod
    async def batch_delete_server_hosts(db: AsyncSession, host_ids: List[int]) -> bool:
        """批量删除主机"""
        return await ServerHostDao.batch_delete_server_hosts(db, host_ids)

    @staticmethod
    async def get_max_host_number(db: AsyncSession, name_prefix: str) -> int:
        """
        获取指定前缀的最大主机序号
        
        Args:
            db: 数据库会话
            name_prefix: 主机名前缀（如'agent'）
            
        Returns:
            最大序号，如果没有匹配的主机则返回0
        """
        import re
        
        # 查询所有以该前缀开头的主机
        hosts = await ServerHostDao.get_hosts_by_name_prefix(db, name_prefix)
        
        # 提取序号并找出最大值
        max_num = 0
        for host in hosts:
            try:
                # 提取 "agent-123" 中的 123
                match = re.search(rf'{re.escape(name_prefix)}-(\d+)$', host.host_name)
                if match:
                    num = int(match.group(1))
                    if num > max_num:
                        max_num = num
            except (ValueError, AttributeError):
                continue
        
        logger.info(f"[批量导入] 主机名前缀 '{name_prefix}' 当前最大序号: {max_num}")
        return max_num
    
    @staticmethod
    async def batch_import_hosts(
        db: AsyncSession,
        import_data: ServerHostBatchImportModel,
        create_by: int,
        dept_id: int,
        auto_init_root: bool = False,
        test_connection: bool = True
    ) -> Dict[str, Any]:
        """
        批量导入主机
        
        Args:
            auto_init_root: 是否自动初始化root账号（适用于ubuntu等有sudo权限的账号）
            test_connection: 是否在导入前测试SSH连接（默认True）
        """
        success_count = 0
        failed_list = []
        host_ids = []  # 记录成功创建的主机ID
        
        # 加密密码
        encrypted_password = PasswordEncryptor.encrypt_password(import_data.ssh_password)
        
        # ✅ 获取当前主机名前缀的最大序号，确保新主机名不重复
        max_num = await ServerHostService.get_max_host_number(db, import_data.name_prefix)
        logger.info(f"[批量导入] 开始导入 {len(import_data.ip_list)} 台主机，主机名从 {import_data.name_prefix}-{max_num + 1} 开始")
        
        for idx, ip in enumerate(import_data.ip_list, 1):
            try:
                # ✅ 生成主机名（使用续接的序号）
                host_name = f"{import_data.name_prefix}-{max_num + idx}"
                
                # ✅ 检查IP是否已存在
                existing_host = await ServerHostDao.get_server_host_by_ip(db, ip)
                if existing_host:
                    failed_list.append({
                        'ip': ip,
                        'host_name': host_name,
                        'reason': 'IP已存在'
                    })
                    continue
                
                # ✅ 检查主机名是否已存在（双重保险）
                existing_host_name = await ServerHostDao.get_server_host_by_name(db, host_name)
                if existing_host_name:
                    failed_list.append({
                        'ip': ip,
                        'host_name': host_name,
                        'reason': f'主机名已存在: {host_name}'
                    })
                    continue
                
                # 如果启用连接测试，先测试SSH连接并检测内网IP
                internal_ip = None
                if test_connection:
                    try:
                        logger.info(f"测试SSH连接: {ip}:{import_data.ssh_port} (用户: {import_data.ssh_username})")
                        test_conn = await SSHConnectionManager.create_connection(
                            ip,
                            import_data.ssh_port,
                            import_data.ssh_username,
                            import_data.ssh_password
                        )
                        # 自动检测内网IP
                        internal_ip = await ServerHostService._detect_internal_ip(test_conn)
                        if internal_ip:
                            logger.info(f"检测到内网IP: {internal_ip}")
                        test_conn.close()
                        logger.info(f"SSH连接测试成功: {ip}")
                    except Exception as e:
                        error_msg = str(e)
                        logger.warning(f"SSH连接测试失败: {ip} - {error_msg}")
                        failed_list.append({
                            'ip': ip,
                            'reason': f'SSH连接失败: {error_msg.split(chr(10))[0]}'  # 只取第一行错误信息
                        })
                        continue
                
                # ✅ 创建主机（使用续接序号的主机名，包含检测到的内网IP）
                host = ServerHost(
                    host_name=host_name,
                    host_ip=ip,
                    internal_ip=internal_ip,  # 保存检测到的内网IP
                    internal_ssh_port=import_data.ssh_port if internal_ip else None,  # 默认使用相同端口
                    ssh_port=import_data.ssh_port,
                    ssh_username=import_data.ssh_username,
                    ssh_password=encrypted_password,
                    agent_path=import_data.agent_path,
                    create_by=create_by,
                    dept_id=dept_id,
                    create_time=datetime.now(),
                    update_time=datetime.now()
                )
                
                created_host = await ServerHostDao.create_server_host(db, host)
                await db.flush()  # 确保获得ID
                
                success_count += 1
                host_ids.append(created_host.id)
                logger.info(f"主机导入成功: {ip} (ID: {created_host.id})")
                
            except Exception as e:
                logger.error(f"导入主机 {ip} 失败: {str(e)}")
                failed_list.append({
                    'ip': ip,
                    'reason': str(e)
                })
        
        result = {
            'success_count': success_count,
            'failed_count': len(failed_list),
            'failed_list': failed_list,
            'host_ids': host_ids
        }
        
        # 如果启用自动初始化root，且使用的是非root账号
        if auto_init_root and import_data.ssh_username != 'root' and host_ids:
            logger.info(f"开始批量初始化root账号，共 {len(host_ids)} 台主机")
            init_results = []
            
            for host_id in host_ids:
                try:
                    init_result = await ServerHostService.init_root_account(db, host_id)
                    init_results.append(init_result)
                except Exception as e:
                    logger.error(f"初始化root账号失败（主机ID: {host_id}）: {e}")
                    init_results.append({
                        'host_id': host_id,
                        'status': 'failed',
                        'message': str(e)
                    })
            
            result['init_root_results'] = init_results
            result['init_root_success'] = sum(1 for r in init_results if r.get('status') == 'success')
        
        return result

    @staticmethod
    async def init_root_account(db: AsyncSession, host_id: int) -> Dict[str, Any]:
        """
        初始化root账号（使用ubuntu账号设置root密码）
        适用于ubuntu账号有sudo权限的情况
        """
        host = await ServerHostDao.get_server_host_by_id(db, host_id)
        if not host:
            raise Exception("主机不存在")
        
        # 只对非root账号执行
        if host.ssh_username == 'root':
            return {
                'status': 'skipped',
                'message': '当前已是root账号，无需初始化'
            }
        
        # 解密密码
        password = PasswordEncryptor.decrypt_password(host.ssh_password)
        
        try:
            logger.info(f"开始初始化root账号: {host.host_ip}")
            
            # 1. 使用当前账号（如ubuntu）连接
            conn = await SSHConnectionManager.create_connection(
                host.host_ip,
                host.ssh_port,
                host.ssh_username,
                password
            )
            
            try:
                # 2. 设置root密码（使用echo和chpasswd，兼容性更好）
                set_password_cmd = f'echo "root:{password}" | sudo chpasswd'
                stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                    conn, set_password_cmd
                )
                
                if exit_status != 0:
                    raise Exception(f"设置root密码失败: {stderr}")
                
                logger.info("root密码设置成功")
                
                # 3. 测试root账号登录
                await asyncio.sleep(1)  # 等待密码生效
                
                root_conn = await SSHConnectionManager.create_connection(
                    host.host_ip,
                    host.ssh_port,
                    'root',
                    password
                )
                root_conn.close()
                
                logger.info("root账号登录测试成功")
                
                # 4. 更新数据库记录
                await ServerHostDao.update_server_host(db, host_id, {
                    'ssh_username': 'root',
                    'connection_status': 'connected',
                    'last_connect_time': datetime.now(),
                    'update_time': datetime.now()
                })
                
                logger.info(f"主机 {host.host_ip} 已自动切换到root账号")
                
                return {
                    'status': 'success',
                    'message': f'root账号初始化成功，已自动切换到root账号登录',
                    'old_username': host.ssh_username,
                    'new_username': 'root'
                }
                
            finally:
                conn.close()
                
        except Exception as e:
            logger.error(f"初始化root账号失败: {e}")
            return {
                'status': 'failed',
                'message': f'初始化失败: {str(e)}'
            }

    @staticmethod
    async def batch_init_root_account(db: AsyncSession, host_ids: List[int]) -> List[Dict[str, Any]]:
        """
        批量初始化root账号
        """
        results = []
        
        for host_id in host_ids:
            try:
                result = await ServerHostService.init_root_account(db, host_id)
                result['host_id'] = host_id
                results.append(result)
            except Exception as e:
                logger.error(f"批量初始化root账号失败（主机ID: {host_id}）: {e}")
                results.append({
                    'host_id': host_id,
                    'status': 'failed',
                    'message': str(e)
                })
        
        return results

    @staticmethod
    def _get_control_server_internal_ip() -> str:
        """
        获取控制端服务器的内网IP
        用于在部署Agent时设置APP_HOST
        """
        try:
            # 通过连接外部地址获取本机IP（通常为内网IP）
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 80))
            local_ip = s.getsockname()[0]
            s.close()
            logger.info(f"获取控制端内网IP: {local_ip}")
            return local_ip
        except Exception as e:
            logger.warning(f"无法获取控制端内网IP: {e}，使用 localhost")
            return "localhost"

    @staticmethod
    async def _detect_internal_ip(conn) -> str:
        """
        自动检测服务器的内网IP地址
        优先使用eth0网卡的IP，如果没有则使用第一个非127.0.0.1的IP
        """
        try:
            # 方法1: 使用 hostname -I（推荐，最简单）
            stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                conn, "hostname -I"
            )
            if exit_status == 0 and stdout.strip():
                # hostname -I 返回所有IP地址，用空格分隔
                # 通常第一个就是内网IP
                ips = stdout.strip().split()
                for ip in ips:
                    if not ip.startswith('127.') and not ip.startswith('172.17.'):  # 排除回环和Docker
                        return ip
            
            # 方法2: 使用 ip addr（更可靠）
            stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                conn, "ip addr show eth0 | grep 'inet ' | awk '{print $2}' | cut -d/ -f1"
            )
            if exit_status == 0 and stdout.strip():
                ip = stdout.strip()
                if ip and not ip.startswith('127.'):
                    return ip
            
            # 方法3: 使用 ifconfig（兼容旧系统）
            stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                conn, "ifconfig eth0 2>/dev/null | grep 'inet ' | awk '{print $2}' | sed 's/addr://'"
            )
            if exit_status == 0 and stdout.strip():
                ip = stdout.strip()
                if ip and not ip.startswith('127.'):
                    return ip
            
            return None
        except Exception as e:
            logger.warning(f"检测内网IP失败: {str(e)}")
            return None
    
    @staticmethod
    async def test_connection(db: AsyncSession, host_id: int) -> Dict[str, Any]:
        """测试SSH连接并自动检测内网IP"""
        host = await ServerHostDao.get_server_host_by_id(db, host_id)
        if not host:
            raise Exception("主机不存在")
        
        # 解密密码
        password = PasswordEncryptor.decrypt_password(host.ssh_password)
        
        # 智能选择IP：优先使用内网IP（如果已设置），否则使用外网IP
        connect_ip = host.internal_ip if host.internal_ip else host.host_ip
        connect_port = host.internal_ssh_port if host.internal_ssh_port else host.ssh_port
        use_internal = bool(host.internal_ip)
        
        start_time = time.time()
        try:
            conn = await SSHConnectionManager.create_connection(
                connect_ip,
                connect_port,
                host.ssh_username,
                password
            )
            
            # 自动检测内网IP（如果还没有）
            internal_ip = None
            if not host.internal_ip:
                internal_ip = await ServerHostService._detect_internal_ip(conn)
                if internal_ip:
                    logger.info(f"✅ 主机 {host.host_name} 检测到内网IP: {internal_ip}")
            
            conn.close()
            
            duration = int((time.time() - start_time) * 1000)
            
            # 确定连接类型
            connection_type = "内网" if use_internal else "外网"
            
            # 更新连接状态、内网IP、延迟和连接类型
            update_data = {
                'connection_status': 'connected',
                'latency': duration,
                'connection_type': connection_type,
                'last_connect_time': datetime.now(),
                'update_time': datetime.now()
            }
            if internal_ip:
                update_data['internal_ip'] = internal_ip
                update_data['internal_ssh_port'] = host.ssh_port  # 默认使用相同端口
            
            await ServerHostDao.update_server_host(db, host_id, update_data)
            
            success_message = f'连接成功（{connection_type}）'
            if internal_ip:
                success_message += f'，检测到内网IP: {internal_ip}'
            
            return {
                'status': 'success',
                'message': success_message,
                'duration': duration,
                'host_id': host_id,
                'host_ip': host.host_ip,
                'host_name': host.host_name,
                'internal_ip': internal_ip or host.internal_ip,  # 返回检测到的或已有的内网IP
                'connection_type': connection_type  # 标识使用的连接类型
            }
            
        except Exception as e:
            # 如果使用内网IP失败，尝试降级到外网IP
            if use_internal:
                logger.warning(f"⚠️ 主机 {host.host_name} 内网IP连接失败，尝试外网IP: {str(e)}")
                try:
                    fallback_start_time = time.time()
                    conn = await SSHConnectionManager.create_connection(
                        host.host_ip,
                        host.ssh_port,
                        host.ssh_username,
                        password
                    )
                    
                    # 检测内网IP（可能网络环境变了）
                    internal_ip = await ServerHostService._detect_internal_ip(conn)
                    if internal_ip and internal_ip != host.internal_ip:
                        logger.info(f"✅ 主机 {host.host_name} 重新检测到内网IP: {internal_ip}")
                    
                    conn.close()
                    
                    duration = int((time.time() - fallback_start_time) * 1000)
                    
                    # 更新连接状态（使用外网降级）
                    update_data = {
                        'connection_status': 'connected',
                        'latency': duration,
                        'connection_type': '外网（降级）',
                        'last_connect_time': datetime.now(),
                        'update_time': datetime.now()
                    }
                    if internal_ip and internal_ip != host.internal_ip:
                        update_data['internal_ip'] = internal_ip
                        update_data['internal_ssh_port'] = host.ssh_port
                    
                    await ServerHostDao.update_server_host(db, host_id, update_data)
                    
                    return {
                        'status': 'success',
                        'message': f'连接成功（外网，内网失败已降级）' + (f'，更新内网IP: {internal_ip}' if internal_ip and internal_ip != host.internal_ip else ''),
                        'duration': duration,
                        'host_id': host_id,
                        'host_ip': host.host_ip,
                        'host_name': host.host_name,
                        'internal_ip': internal_ip or host.internal_ip,
                        'connection_type': '外网（降级）'
                    }
                except Exception as fallback_error:
                    logger.error(f"❌ 主机 {host.host_name} 外网IP连接也失败: {str(fallback_error)}")
                    # 更新连接状态为错误
                    await ServerHostDao.update_server_host(db, host_id, {
                        'connection_status': 'error',
                        'update_time': datetime.now()
                    })
                    return {
                        'status': 'failed',
                        'message': f'内网和外网均连接失败 - 内网: {str(e)}, 外网: {str(fallback_error)}',
                        'duration': int((time.time() - start_time) * 1000),
                        'host_id': host_id,
                        'host_ip': host.host_ip,
                        'host_name': host.host_name
                    }
            else:
                # 使用外网IP失败（首次连接）
                await ServerHostDao.update_server_host(db, host_id, {
                    'connection_status': 'error',
                    'update_time': datetime.now()
                })
                
                return {
                    'status': 'failed',
                    'message': str(e),
                    'duration': int((time.time() - start_time) * 1000),
                    'host_id': host_id,
                    'host_ip': host.host_ip,
                    'host_name': host.host_name
                }
    
    @staticmethod
    async def batch_test_connection(db: AsyncSession, host_ids: List[int]) -> List[Dict[str, Any]]:
        """批量测试连接"""
        results = []
        
        # 并发测试所有主机
        tasks = []
        for host_id in host_ids:
            tasks.append(ServerHostService.test_connection(db, host_id))
        
        # 等待所有测试完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理异常结果
        final_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                host_id = host_ids[i]
                final_results.append({
                    'status': 'failed',
                    'message': str(result),
                    'duration': 0,
                    'host_id': host_id,
                    'host_ip': '',
                    'host_name': ''
                })
            else:
                final_results.append(result)
        
        return final_results

    @staticmethod
    async def execute_command(
        db: AsyncSession,
        execute_data: CommandExecuteModel,
        create_by: int
    ) -> List[CommandExecuteResultModel]:
        """批量执行命令（并发执行）"""
        hosts = await ServerHostDao.get_hosts_by_ids(db, execute_data.host_ids)
        
        # ✅ 提取主机快照，避免 lazy loading
        host_snapshots = [
            {
                'id': host.id,
                'host_ip': host.host_ip,
                'host_name': host.host_name
            }
            for host in hosts
        ]
        
        # ✅ 使用信号量限制并发数，避免资源耗尽和会话冲突
        semaphore = asyncio.Semaphore(5)  # 一次最多5个并发任务
        
        async def execute_single_host(snapshot):
            """在单个主机上执行命令，使用独立的数据库会话"""
            async with semaphore:
                # ✅ 每个任务创建独立的数据库会话
                async with AsyncSessionLocal() as task_db:
                    # ✅ 重新查询 host 对象，避免使用 expired 的对象
                    host = await ServerHostDao.get_server_host_by_id(task_db, snapshot['id'])
                    if not host:
                        raise Exception(f"未找到主机 ID: {snapshot['id']}")
                    
                    return await ServerHostService._execute_command_on_host(
                        task_db, host, execute_data.command, execute_data.command_type, create_by
                )
        
        # 并发执行所有主机的命令
        tasks = [execute_single_host(snapshot) for snapshot in host_snapshots]
        
        # 等待所有任务完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理异常结果
        final_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                snapshot = host_snapshots[i]
                final_results.append(
                    CommandExecuteResultModel(
                        host_id=snapshot['id'],
                        host_ip=snapshot['host_ip'],
                        host_name=snapshot['host_name'],
                        status='failed',
                        error=str(result),
                        duration=0
                    )
                )
            else:
                final_results.append(result)
        
        return final_results

    @staticmethod
    async def _execute_command_on_host(
        db: AsyncSession,
        host: ServerHost,
        command: str,
        command_type: str,
        create_by: int
    ) -> CommandExecuteResultModel:
        """在单个主机上执行命令"""
        start_time = time.time()
        
        # 创建命令日志
        log = ServerCommandLog(
            host_id=host.id,
            host_ip=host.host_ip,
            command=command,
            command_type=command_type,
            status='pending',
            create_by=create_by,
            create_time=datetime.now()
        )
        log = await ServerCommandLogDao.create_command_log(db, log)
        
        try:
            # 解密密码
            password = PasswordEncryptor.decrypt_password(host.ssh_password)
            
            # 创建SSH连接
            conn = await SSHConnectionManager.create_connection(
                host.host_ip,
                host.ssh_port,
                host.ssh_username,
                password
            )
            
            try:
                # 执行命令
                stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                    conn, command
                )
                
                duration = int((time.time() - start_time) * 1000)
                
                # 更新日志
                await ServerCommandLogDao.update_command_log(db, log.id, {
                    'result': stdout,
                    'error': stderr if stderr else None,
                    'status': 'success' if exit_status == 0 else 'failed',
                    'execute_time': datetime.now(),
                    'duration': duration
                })
                
                return CommandExecuteResultModel(
                    host_id=host.id,
                    host_ip=host.host_ip,
                    host_name=host.host_name,
                    status='success' if exit_status == 0 else 'failed',
                    result=stdout,
                    error=stderr if stderr else None,
                    duration=duration
                )
                
            finally:
                conn.close()
                
        except Exception as e:
            duration = int((time.time() - start_time) * 1000)
            error_msg = str(e)
            
            # 更新日志
            await ServerCommandLogDao.update_command_log(db, log.id, {
                'error': error_msg,
                'status': 'failed',
                'execute_time': datetime.now(),
                'duration': duration
            })
            
            return CommandExecuteResultModel(
                host_id=host.id,
                host_ip=host.host_ip,
                host_name=host.host_name,
                status='failed',
                error=error_msg,
                duration=duration
            )

    @staticmethod
    async def deploy_agent(
        db: AsyncSession,
        deploy_data: AgentDeployModel,
        create_by: int
    ) -> List[CommandExecuteResultModel]:
        """批量部署Agent"""
        hosts = await ServerHostDao.get_hosts_by_ids(db, deploy_data.host_ids)
        
        # ✅ 在 commit 前提取所有需要的数据，避免 lazy loading
        # 只保存必要的基本信息用于错误处理，子任务会重新查询完整对象
        host_snapshots = [
            {
                'id': host.id,
                'host_ip': host.host_ip,
                'host_name': host.host_name
            }
            for host in hosts
        ]
        
        # 更新所有主机状态为部署中
        for snapshot in host_snapshots:
            await ServerHostDao.update_server_host(db, snapshot['id'], {
                'deploy_status': 'deploying',
                'deploy_error': None,
                'update_time': datetime.now()
            })
        
        # ✅ 立即提交事务，释放锁，避免子任务被阻塞
        await db.commit()
        
        # ✅ 使用信号量限制并发数，避免资源耗尽和会话冲突
        # 一次最多3个并发部署任务
        semaphore = asyncio.Semaphore(3)
        
        async def deploy_single_host(snapshot):
            """部署单个主机，确保在独立的异步上下文中运行"""
            async with semaphore:
                # ✅ 每个任务创建独立的数据库会话
                # 使用上下文管理器，确保在正确的异步上下文中
                async with AsyncSessionLocal() as task_db:
                    # ✅ 重新查询 host 对象，避免使用 expired 的对象
                    # 主会话的 host 对象在 commit 后已经 expired，访问属性会触发 lazy loading
                    host = await ServerHostDao.get_server_host_by_id(task_db, snapshot['id'])
                    if not host:
                        raise Exception(f"未找到主机 ID: {snapshot['id']}")
                    
                    result = await ServerHostService._deploy_agent_on_host(
                        task_db, host, deploy_data.agent_file_path, create_by
                    )
                    return result
        
        # ✅ 并发部署（但有限制）
        # gather 会自动创建任务，确保每个任务在独立的异步上下文中运行
        logger.info(f"🚀 开始并发部署 {len(host_snapshots)} 台主机（最多3个并发）")
        host_list = [f"{s['host_name']}({s['host_ip']})" for s in host_snapshots]
        logger.info(f"主机列表: {host_list}")
        
        results = await asyncio.gather(
            *[deploy_single_host(snapshot) for snapshot in host_snapshots],
            return_exceptions=True
        )
        
        logger.info(f"✅ 所有部署任务完成")
        
        # 处理结果（任务内部已经更新了数据库，这里只需要处理异常情况）
        final_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                # 异常处理（通常是连接失败等严重错误）
                error_msg = str(result)
                snapshot = host_snapshots[i]
                final_results.append(CommandExecuteResultModel(
                    host_id=snapshot['id'],
                    host_ip=snapshot['host_ip'],
                    host_name=snapshot['host_name'],
                    status='failed',
                    error=error_msg,
                    duration=0
                ))
                # ✅ 异常情况下，用主会话再次更新状态（任务可能未完成更新）
                try:
                    await ServerHostDao.update_server_host(db, snapshot['id'], {
                    'deploy_status': 'deploy_failed',
                    'deploy_error': error_msg,
                    'update_time': datetime.now()
                })
                except Exception as update_err:
                    logger.error(f"更新主机状态失败 {snapshot['id']}: {update_err}")
            else:
                final_results.append(result)
        
        return final_results

    @staticmethod
    async def _deploy_agent_on_host(
        db: AsyncSession,
        host: ServerHost,
        agent_file_path: str,
        create_by: int
    ) -> CommandExecuteResultModel:
        """在单个主机上部署Agent"""
        start_time = time.time()
        
        # ✅ 在函数最开始就提前获取所有host属性，避免后续lazy loading触发greenlet错误
        agent_path = host.agent_path
        host_id = host.id
        host_ip = host.host_ip
        host_name = host.host_name
        ssh_password = host.ssh_password
        ssh_port = host.ssh_port
        ssh_username = host.ssh_username
        
        logger.info(f"⏳ [{host_name}] 开始部署...")
        
        try:
            # 解密密码
            password = PasswordEncryptor.decrypt_password(ssh_password)
            
            # 创建SSH连接
            conn = await SSHConnectionManager.create_connection(
                host_ip,
                ssh_port,
                ssh_username,
                password
            )
            
            try:
                temp_file = f"/tmp/agent_{int(time.time())}.tar.gz"
                
                # ✅ 判断是 URL 还是本地文件路径
                if agent_file_path.startswith(('http://', 'https://')):
                    # 1. 使用 wget 从控制端下载（比 SFTP 更快）
                    logger.info(f"⬇️  [{host_name}] 开始下载Agent文件...")
                    download_cmd = f"wget --quiet --timeout=60 --tries=3 -O {temp_file} '{agent_file_path}'"
                    stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                        conn, download_cmd, timeout=180
                    )
                    
                    if exit_status != 0:
                        # 如果 wget 失败，尝试 curl
                        logger.warning(f"[{host_name}] wget失败，尝试curl")
                        download_cmd = f"curl --silent --fail --connect-timeout 60 --max-time 180 -o {temp_file} '{agent_file_path}'"
                        stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                            conn, download_cmd, timeout=200
                        )
                        
                        if exit_status != 0:
                            raise Exception(f"下载Agent文件失败 (wget/curl都失败): {stderr}")
                    
                    logger.success(f"✅ [{host_name}] 文件下载完成")
                else:
                    # 1. 兼容旧方式：使用 SFTP 上传本地文件
                    logger.info(f"⬆️  [{host_name}] 使用SFTP上传Agent文件...")
                    await SSHConnectionManager.upload_file(
                        conn, agent_file_path, temp_file
                    )
                    logger.success(f"✅ [{host_name}] 文件上传完成")
                
                # ✅ 2. 使用 sudo bash -c 执行完整的部署流程（避免权限问题）
                logger.info(f"🚀 [{host_name}] 开始部署...")
                
                # ✅ 去掉 agent_path 末尾的斜杠（如果有）
                agent_path_clean = agent_path.rstrip('/')
                
                backup_dir = f"/tmp/agent_backup_{int(time.time())}"
                
                # ✅ 获取控制端内网IP（用于设置APP_HOST）
                control_server_ip = ServerHostService._get_control_server_internal_ip()
                logger.info(f"🔧 [{host_name}] 控制端内网IP: {control_server_ip}")
                
                # ✅ 组合所有命令为一个完整的bash脚本，用sudo执行
                # 🎯 关键改进：在 /tmp 解压，检查完整性后再移动到目标目录
                temp_extract_dir = f"/tmp/agent_extract_{int(time.time())}"
                
                deploy_commands = f"""
# ✅ 设置完整的环境变量（确保命令可以找到）
export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
export LANG=en_US.UTF-8
export LC_ALL=en_US.UTF-8

set -e  # ⚠️ 任何命令失败立即退出
set -x  # 📝 输出每个命令的执行（调试用）

echo "🛑 停止服务..."
/usr/bin/systemctl stop goofish-agent 2>/dev/null || true

echo "🧹 清理进程..."
/usr/bin/pkill -9 -f "python.*main.py" 2>/dev/null || true
/usr/bin/pkill -9 -f "chrome" 2>/dev/null || true
/usr/bin/pkill -9 -f "chromium" 2>/dev/null || true

echo "💾 备份浏览器配置文件夹..."
if [ -d "/opt/user_data/default" ]; then
    browser_backup_timestamp=$(date +%Y%m%d_%H%M%S)
    browser_backup_dir="/opt/user_data/default_backup_$browser_backup_timestamp"
    /bin/mv "/opt/user_data/default" "$browser_backup_dir" 2>/dev/null || true
    if [ -d "$browser_backup_dir" ]; then
        echo "✅ 已备份浏览器配置文件夹到: $browser_backup_dir"
    fi
fi

echo "🔓 清理Chrome锁定文件（如果存在）..."
/bin/rm -f /opt/user_data/default/SingletonLock 2>/dev/null || true
/bin/rm -f /opt/user_data/default/SingletonSocket 2>/dev/null || true
/bin/rm -f /opt/user_data/default/SingletonCookie 2>/dev/null || true

echo "📦 在 /tmp 解压压缩包..."
/bin/mkdir -p {temp_extract_dir}
cd {temp_extract_dir}
/bin/tar -xzf {temp_file}

echo "✅ 检查解压结果..."
if [ ! -f "main.py" ]; then
    echo "❌ ERROR: main.py 不存在！"
    /bin/ls -la {temp_extract_dir} || true
    exit 1
fi

if [ ! -f "deploy_agent.sh" ]; then
    echo "❌ ERROR: deploy_agent.sh 不存在！"
    /bin/ls -la {temp_extract_dir} || true
    exit 1
fi

echo "✅ 解压成功，文件完整"
/bin/ls -la {temp_extract_dir} || true

echo "💾 备份旧配置文件..."
/bin/mkdir -p {backup_dir}
if [ -d "{agent_path_clean}" ]; then
    [ -f "{agent_path_clean}/config.json" ] && /bin/cp "{agent_path_clean}/config.json" {backup_dir}/ 2>/dev/null || true
    [ -f "{agent_path_clean}/state.json" ] && /bin/cp "{agent_path_clean}/state.json" {backup_dir}/ 2>/dev/null || true
    [ -f "{agent_path_clean}/.env" ] && /bin/cp "{agent_path_clean}/.env" {backup_dir}/ 2>/dev/null || true
fi

echo "🗑️ 删除旧目录: {agent_path_clean}"
/bin/rm -rf {agent_path_clean}

echo "📁 移动解压后的文件到目标目录..."
/bin/mv {temp_extract_dir} {agent_path_clean}

echo "♻️ 恢复配置文件..."
if [ -d "{backup_dir}" ]; then
    [ -f "{backup_dir}/config.json" ] && /bin/cp {backup_dir}/config.json {agent_path_clean}/ 2>/dev/null || true
    [ -f "{backup_dir}/state.json" ] && /bin/cp {backup_dir}/state.json {agent_path_clean}/ 2>/dev/null || true
    [ -f "{backup_dir}/.env" ] && /bin/cp {backup_dir}/.env {agent_path_clean}/ 2>/dev/null || true
    /bin/rm -rf {backup_dir}
fi

echo "🔧 修改.env文件中的APP_HOST为控制端内网IP..."
if [ -f "{agent_path_clean}/.env" ]; then
    # 如果.env文件中存在APP_HOST，则更新它；如果不存在，则添加它
    if /bin/grep -q "^APP_HOST=" "{agent_path_clean}/.env"; then
        # 使用sed替换APP_HOST的值
        /bin/sed -i "s|^APP_HOST=.*|APP_HOST={control_server_ip}|" "{agent_path_clean}/.env"
        echo "✅ 已更新APP_HOST={control_server_ip}"
    else
        # 如果不存在，则追加到文件末尾
        echo "APP_HOST={control_server_ip}" >> "{agent_path_clean}/.env"
        echo "✅ 已添加APP_HOST={control_server_ip}"
    fi
else
    # 如果.env文件不存在，创建一个新的
    echo "APP_HOST={control_server_ip}" > "{agent_path_clean}/.env"
    echo "✅ 已创建.env文件并设置APP_HOST={control_server_ip}"
fi

echo "✅ 文件准备完成（root用户，无需chown）"

echo "🧹 清理临时文件..."
/bin/rm -f {temp_file}

echo "✅ 准备阶段完成！"
/bin/ls -la {agent_path_clean} || true

# ✅ 显式返回成功状态（重要！）
exit 0
"""
                
                # ✅ 使用 SFTP 上传脚本文件（最可靠的方式）
                prep_script_file = f"/tmp/prep_script_{int(time.time())}.sh"
                
                logger.info(f"📦 [{host_name}] 准备部署环境...")
                logger.debug(f"脚本文件: {prep_script_file}")
                logger.debug(f"脚本内容长度: {len(deploy_commands)} 字节")
                
                try:
                    # 第一步：通过 SFTP 上传脚本文件
                    async with conn.start_sftp_client() as sftp:
                        # 写入脚本内容到远程文件
                        async with sftp.open(prep_script_file, 'w') as f:
                            await f.write(deploy_commands)
                        
                        # 设置执行权限
                        await sftp.chmod(prep_script_file, 0o755)
                    
                    logger.debug(f"✅ [{host_name}] 脚本文件已上传")
                    
                    # 第二步：执行脚本并清理
                    exec_cmd = f"sudo bash {prep_script_file}; EXIT_CODE=$?; rm -f {prep_script_file}; exit $EXIT_CODE"
                    
                    prep_stdout, prep_stderr, prep_exit_status = await SSHConnectionManager.execute_command(
                        conn, exec_cmd,
                        timeout=300  # 准备工作最多5分钟
                    )
                    logger.info(f"📦 [{host_name}] 准备阶段执行完成，退出码: {prep_exit_status}")
                    logger.debug(f"准备阶段 stdout 长度: {len(prep_stdout) if prep_stdout else 0}")
                    logger.debug(f"准备阶段 stderr 长度: {len(prep_stderr) if prep_stderr else 0}")
                    
                    # ✅ 由于 set -x 会把调试信息输出到 stderr，所以 stderr 不为空是正常的
                    # 只要退出码是 0 就说明成功
                    if prep_stderr:
                        logger.debug(f"准备阶段 stderr 预览: {prep_stderr[:200] if prep_stderr else 'None'}")
                except Exception as e:
                    logger.error(f"❌ [{host_name}] 准备阶段执行异常: {str(e)}")
                    raise
                
                # ✅ 只判断退出码，stderr 不重要（因为 set -x 的调试输出会到 stderr）
                if prep_exit_status != 0:
                    logger.error(f"❌ [{host_name}] 准备环境失败，退出码: {prep_exit_status}")
                    logger.error(f"完整 stderr 输出:")
                    logger.error(prep_stderr)
                    raise Exception(f"准备环境失败（退出码: {prep_exit_status}），请查看日志")
                
                logger.success(f"✅ [{host_name}] 环境准备完成")
                
                # ✅ 再单独执行deploy_agent.sh（确保完整执行）
                deploy_script_cmd = f"cd {agent_path_clean} && sudo bash deploy_agent.sh"
                
                logger.info(f"🚀 [{host_name}] 执行部署脚本...")
                exec_cmd = deploy_script_cmd
                
                deploy_stdout, deploy_stderr, deploy_exit_status = await SSHConnectionManager.execute_command(
                    conn, exec_cmd,
                    timeout=1800  # 30分钟超时
                )
                
                # ✅ 使用root用户运行，无需chown操作
                
                # ✅ 优化输出：只显示关键信息
                if deploy_exit_status == 0:
                    # 从部署输出中提取关键信息
                    summary_lines = []
                    summary_lines.append("✅ 部署完成！")
                    summary_lines.append("")
                    
                    # 提取服务状态
                    if "服务状态:" in deploy_stdout:
                        for line in deploy_stdout.split('\n'):
                            if "服务状态:" in line or "服务名称:" in line:
                                summary_lines.append(f"[INFO] {line.strip()}")
                    
                    # 提取部署报告路径
                    if "部署报告" in deploy_stdout or "deploy_report.txt" in deploy_stdout:
                        for line in deploy_stdout.split('\n'):
                            if "部署报告" in line or "deploy_report.txt" in line:
                                summary_lines.append(f"[INFO] {line.strip()}")
                    
                    # 提取部署日志路径
                    if "部署日志" in deploy_stdout or "deploy.log" in deploy_stdout:
                        for line in deploy_stdout.split('\n'):
                            if "部署日志" in line or "deploy.log" in line:
                                summary_lines.append(f"[INFO] {line.strip()}")
                    
                    # 如果没有提取到关键信息，显示最后几行
                    if len(summary_lines) <= 2:
                        deploy_lines = deploy_stdout.split('\n')
                        # 取最后10行
                        summary_lines.extend([f"[INFO] {line.strip()}" for line in deploy_lines[-10:] if line.strip()])
                    
                    stdout = '\n'.join(summary_lines)
                else:
                    # 失败时显示完整输出
                    stdout = f"""
========================================
📦 准备阶段输出
========================================
{prep_stdout}

========================================
🚀 部署脚本输出
========================================
{deploy_stdout}
"""
                
                stderr = f"{prep_stderr}\n{deploy_stderr}" if prep_stderr or deploy_stderr else None
                exit_status = deploy_exit_status
                
                duration = int((time.time() - start_time) * 1000)
                
                # 根据执行结果更新部署状态
                deploy_status = 'deployed' if exit_status == 0 else 'deploy_failed'
                deploy_error = stderr if exit_status != 0 and stderr else None
                
                # 更新主机部署状态和时间
                await ServerHostDao.update_server_host(db, host_id, {
                    'last_deploy_time': datetime.now(),
                    'deploy_status': deploy_status,
                    'deploy_error': deploy_error,
                    'agent_status': 'stopped' if exit_status == 0 else 'unknown',
                    'update_time': datetime.now()
                })
                
                # 创建日志
                log = ServerCommandLog(
                    host_id=host_id,
                    host_ip=host_ip,
                    command=f"部署Agent到 {agent_path}",
                    command_type='deploy',
                    result=stdout,
                    error=stderr if stderr else None,
                    status='success' if exit_status == 0 else 'failed',
                    execute_time=datetime.now(),
                    duration=duration,
                    create_by=create_by,
                    create_time=datetime.now()
                )
                await ServerCommandLogDao.create_command_log(db, log)
                
                # ✅ 提交事务
                await db.commit()
                
                return CommandExecuteResultModel(
                    host_id=host_id,
                    host_ip=host_ip,
                    host_name=host_name,
                    status='success' if exit_status == 0 else 'failed',
                    result=stdout,
                    error=stderr if stderr else None,
                    duration=duration
                )
                
            finally:
                conn.close()
                
        except Exception as e:
            duration = int((time.time() - start_time) * 1000)
            error_msg = str(e)
            
            # ✅ 记录异常到日志
            logger.error(f"❌ [{host_name}] 部署失败: {error_msg}")
            logger.exception(e)  # 输出完整堆栈
            
            # 更新为部署失败
            await ServerHostDao.update_server_host(db, host_id, {
                'deploy_status': 'deploy_failed',
                'deploy_error': error_msg,
                'update_time': datetime.now()
            })
            
            # ✅ 提交事务
            await db.commit()
            
            # 创建失败日志
            log = ServerCommandLog(
                host_id=host_id,
                host_ip=host_ip,
                command=f"部署Agent到 {agent_path}",
                command_type='deploy',
                error=error_msg,
                status='failed',
                execute_time=datetime.now(),
                duration=duration,
                create_by=create_by,
                create_time=datetime.now()
            )
            await ServerCommandLogDao.create_command_log(db, log)
            
            # ✅ 提交事务
            await db.commit()
            
            return CommandExecuteResultModel(
                host_id=host_id,
                host_ip=host_ip,
                host_name=host_name,
                status='failed',
                error=error_msg,
                duration=duration
            )

    @staticmethod
    async def control_agent(
        db: AsyncSession,
        control_data: AgentControlModel,
        create_by: int
    ) -> List[CommandExecuteResultModel]:
        """批量控制Agent（启动/停止/重启）"""
        hosts = await ServerHostDao.get_hosts_by_ids(db, control_data.host_ids)
        results = []
        
        # ✅ 使用 systemctl 管理服务
        action_commands = {
            'start': 'sudo systemctl start goofish-agent',
            'stop': 'sudo systemctl stop goofish-agent',
            'restart': 'sudo systemctl restart goofish-agent',
            'status': 'sudo systemctl status goofish-agent --no-pager'
        }
        
        command = action_commands.get(control_data.action)
        if not command:
            raise Exception(f"不支持的操作: {control_data.action}")
        
        for host in hosts:
            result = await ServerHostService._execute_command_on_host(
                db, host, command, control_data.action, create_by
            )
            
            # ✅ 对于 status 操作，即使命令返回非0，只要能从输出中解析出状态，就算成功
            if control_data.action == 'status':
                output = result.result or ''
                # 从输出中判断服务状态
                if 'active (running)' in output.lower():
                    new_status = 'running'
                    # 创建新的结果对象，状态为成功（即使命令返回非0）
                    result = CommandExecuteResultModel(
                        host_id=result.host_id,
                        host_ip=result.host_ip,
                        host_name=result.host_name,
                        status='success',
                        result=result.result,
                        error=result.error,
                        duration=result.duration
                    )
                elif 'inactive (dead)' in output.lower() or 'stopped' in output.lower():
                    new_status = 'stopped'
                    # 创建新的结果对象，状态为成功（即使命令返回非0）
                    result = CommandExecuteResultModel(
                        host_id=result.host_id,
                        host_ip=result.host_ip,
                        host_name=result.host_name,
                        status='success',
                        result=result.result,
                        error=result.error,
                        duration=result.duration
                    )
                elif 'active (exited)' in output.lower():
                    new_status = 'stopped'
                    result = CommandExecuteResultModel(
                        host_id=result.host_id,
                        host_ip=result.host_ip,
                        host_name=result.host_name,
                        status='success',
                        result=result.result,
                        error=result.error,
                        duration=result.duration
                    )
                else:
                    new_status = 'unknown'
                    # 如果无法解析状态，保持原状态（可能是failed）
                
                # 只要能解析出状态，就更新数据库
                if new_status != 'unknown':
                    await ServerHostDao.update_server_host(db, host.id, {
                        'agent_status': new_status,
                        'update_time': datetime.now()
                    })
            elif result.status == 'success':
                # 其他操作（start/stop/restart）只有成功时才更新状态
                new_status = 'running' if control_data.action in ['start', 'restart'] else 'stopped'
                await ServerHostDao.update_server_host(db, host.id, {
                    'agent_status': new_status,
                    'update_time': datetime.now()
                })
            
            results.append(result)
        
        await db.commit()
        return results

    @staticmethod
    async def get_agent_log(
        db: AsyncSession,
        query: AgentLogQueryModel
    ) -> Dict[str, Any]:
        """获取Agent日志"""
        host = await ServerHostDao.get_server_host_by_id(db, query.host_id)
        if not host:
            raise Exception("主机不存在")
        
        try:
            # 解密密码
            password = PasswordEncryptor.decrypt_password(host.ssh_password)
            
            # 创建SSH连接
            conn = await SSHConnectionManager.create_connection(
                host.host_ip,
                host.ssh_port,
                host.ssh_username,
                password
            )
            
            try:
                # 确定日志文件路径
                if query.log_date:
                    log_file = f"{host.agent_path}/logs/{query.log_date}.log"
                else:
                    # 获取最新的日志文件
                    log_file = f"{host.agent_path}/logs/$(ls -t {host.agent_path}/logs/*.log | head -1 | xargs basename)"
                
                # 读取日志内容（最后N行）
                if query.tail_lines:
                    command = f"tail -n {query.tail_lines} {log_file}"
                else:
                    command = f"cat {log_file}"
                
                stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                    conn, command
                )
                
                if exit_status == 0:
                    return {
                        'host_id': host.id,
                        'host_ip': host.host_ip,
                        'log_content': stdout,
                        'log_file': log_file
                    }
                else:
                    raise Exception(stderr or "读取日志失败")
                
            finally:
                conn.close()
                
        except Exception as e:
            logger.error(f"获取日志失败: {str(e)}")
            raise Exception(f"获取日志失败: {str(e)}")

    @staticmethod
    async def get_agent_log_dates(db: AsyncSession, host_id: int) -> List[str]:
        """获取可用的日志日期列表"""
        host = await ServerHostDao.get_server_host_by_id(db, host_id)
        if not host:
            raise Exception("主机不存在")
        
        try:
            # 解密密码
            password = PasswordEncryptor.decrypt_password(host.ssh_password)
            
            # 创建SSH连接
            conn = await SSHConnectionManager.create_connection(
                host.host_ip,
                host.ssh_port,
                host.ssh_username,
                password
            )
            
            try:
                # 列出日志文件
                command = f"ls {host.agent_path}/logs/*.log 2>/dev/null | xargs -n 1 basename | sed 's/.log//'"
                stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                    conn, command
                )
                
                if exit_status == 0 and stdout:
                    dates = [date.strip() for date in stdout.strip().split('\n') if date.strip()]
                    return sorted(dates, reverse=True)  # 降序排列
                else:
                    return []
                
            finally:
                conn.close()
                
        except Exception as e:
            logger.error(f"获取日志日期列表失败: {str(e)}")
            return []

    @staticmethod
    async def check_agent_status(db: AsyncSession, host_id: int) -> Dict[str, Any]:
        """检查Agent运行状态"""
        host = await ServerHostDao.get_server_host_by_id(db, host_id)
        if not host:
            raise Exception("主机不存在")
        
        try:
            # 解密密码
            password = PasswordEncryptor.decrypt_password(host.ssh_password)
            
            # 创建SSH连接
            conn = await SSHConnectionManager.create_connection(
                host.host_ip,
                host.ssh_port,
                host.ssh_username,
                password
            )
            
            try:
                # 检查进程是否运行
                command = f"ps aux | grep '[m]ain.py' | grep '{host.agent_path}'"
                stdout, stderr, exit_status = await SSHConnectionManager.execute_command(
                    conn, command
                )
                
                is_running = bool(stdout.strip())
                new_status = 'running' if is_running else 'stopped'
                
                # 更新状态
                await ServerHostDao.update_server_host(db, host.id, {
                    'agent_status': new_status,
                    'update_time': datetime.now()
                })
                
                return {
                    'host_id': host.id,
                    'agent_status': new_status,
                    'is_running': is_running,
                    'process_info': stdout.strip() if is_running else None
                }
                
            finally:
                conn.close()
                
        except Exception as e:
            logger.error(f"检查Agent状态失败: {str(e)}")
            # 更新为错误状态
            await ServerHostDao.update_server_host(db, host.id, {
                'agent_status': 'error',
                'update_time': datetime.now()
            })
            raise Exception(f"检查Agent状态失败: {str(e)}")

